package org.pupil.rpc.poll;

import io.netty.channel.ChannelHandlerContext;
import org.pupil.rpc.common.RpcServiceNameBuilder;
import org.pupil.rpc.common.constants.MsgStatus;
import org.pupil.rpc.common.constants.MsgType;
import org.pupil.rpc.config.RpcServiceConfig;
import org.pupil.rpc.protocol.MsgHeader;
import org.pupil.rpc.protocol.RpcProtocol;
import org.pupil.rpc.protocol.RpcRequest;
import org.pupil.rpc.protocol.RpcResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cglib.reflect.FastClass;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;


/**
 * 线程池工厂
 */
public class ThreadPollFactory {

    private static Logger logger = LoggerFactory.getLogger(ThreadPollFactory.class);

    // 慢处理任务线程池
    private static ThreadPoolExecutor slowPoll;

    // 快处理任务线程池
    private static ThreadPoolExecutor fastPoll;

    // 慢任务集合, volatile 标志为易变
    private static volatile ConcurrentHashMap<String, AtomicInteger> slowTaskMap = new ConcurrentHashMap();

    // 获取 JVM 可用的处理器（CPU 核心）数量
    private static int corSize = Runtime.getRuntime().availableProcessors();

    // 缓存服务
    private static Map<String, Object> rpcServiceMap = RpcServiceConfig.rpcServiceMap;

    static {
        slowPoll = new ThreadPoolExecutor(corSize / 2,    // 线程池的核心线程数
                corSize,    // 线程池的最大线程数
                60L,    // 非核心线程的空闲最长等待时间
                TimeUnit.SECONDS,   // 时间单位，秒
                new LinkedBlockingDeque<>(2000),    // 创建双向链表阻塞队列，适用于生产者-消费者模式，初始容量为2000
                r -> {  //创建线程
                    Thread thread = new Thread(r);
                    thread.setName("slow poll-" + r.hashCode());
                    thread.setDaemon(true); //开启线程守护
                    return thread;
                });

        fastPoll = new ThreadPoolExecutor(corSize, corSize * 2, 60L,
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(1000),    // 创建双向链表阻塞队列，适用于生产者-消费者模式，初始容量为1000
                r -> {
                    Thread thread = new Thread(r);
                    thread.setName("fast poll-" + r.hashCode());
                    thread.setDaemon(true);     // 开启线程守护
                    return thread;
                });

        startClearMonitor(); //启动一个监控任务清理线程
    }

    private ThreadPollFactory() {
    }

    /**
     * 清理慢请求
     */
    private static void startClearMonitor() {
        Executors.newSingleThreadScheduledExecutor() // 创建一个单线程的调度线程池
                .scheduleWithFixedDelay(() -> { // 设置周期性的任务
                            slowTaskMap.clear(); // 执行的任务
                        }, 5, // 第一次执行任务之前的延迟时间
                        5, // 连续执行任务之间的间隔时间
                        TimeUnit.MINUTES); // 时间单位为分钟
    }

    /**
     * 提交请求
     *
     * @param ctx
     * @param protocol
     */
    public static void submitRequest(ChannelHandlerContext ctx, RpcProtocol<RpcRequest> protocol) {
        final RpcRequest request = protocol.getBody();
        String key = request.getClassName() + request.getMethodName() + request.getServiceVersion();
        ThreadPoolExecutor poll = fastPoll;
        if (slowTaskMap.containsKey(key) && slowTaskMap.get(key).intValue() >= 10) {
            poll = slowPoll;
        }

        poll.submit(() -> {
            RpcProtocol<RpcResponse> resRpcProtocol = new RpcProtocol<>();
            final MsgHeader header = protocol.getHeader();
            RpcResponse response = new RpcResponse();
            // 开始时间
            long startTime = System.currentTimeMillis();

            try {
                // 调用方法
                final Object result = submit(protocol);
                response.setData(result);
                response.setDataClass(result == null ? null : result.getClass());
                header.setStatus((byte) MsgStatus.SUCCESS.ordinal());
            } catch (Exception e) {
                // 业务执行失败则将异常返回
                header.setStatus((byte) MsgStatus.FAILED.ordinal());
                response.setException(e);
                logger.error("process request {} error", header.getRequestId(), e);
            } finally {
                long cost = System.currentTimeMillis() - startTime;
                System.out.println("cost time:" + cost);
                if (cost > 1000) {
                    // key存在返回当前值，key不存在返回null
                    final AtomicInteger timeOutCount = slowTaskMap.putIfAbsent(key, new AtomicInteger(1));
                    if (timeOutCount != null) {
                        // 计数器加一
                        timeOutCount.incrementAndGet();
                    }
                }
            }
            resRpcProtocol.setHeader(header);
            resRpcProtocol.setBody(response);
            logger.info("执行成功: {},{},{},{}", Thread.currentThread().getName(), request.getClassName(), request.getMethodName(), request.getServiceVersion());
            // 传入下一个处理器处理
            ctx.fireChannelRead(resRpcProtocol);
        });
    }

    private static Object submit(RpcProtocol<RpcRequest> protocol) throws Exception{
        MsgHeader header = protocol.getHeader();
        header.setMsgType((byte) MsgType.RESPONSE.ordinal());
        final RpcRequest request = protocol.getBody();
        // 执行具体业务
        return handle(request);
    }

    /**
     * 调用方法，获取返回值
     *
     * @param request
     * @return
     * @throws Exception
     */
    private static Object handle(RpcRequest request) throws Exception {
        // 获取服务名
        String serviceKey = RpcServiceNameBuilder.buildServiceKey(request.getClassName(), request.getServiceVersion());
        // 获取服务的实例类
        Object serviceBean = rpcServiceMap.get(serviceKey);

        if (serviceBean == null) {
            throw new RuntimeException(String.format("service not exist: %s:%s", request.getClassName(), request.getMethodName()));
        }

        // 获取服务提供方信息并创建
        Class<?> serviceClass = serviceBean.getClass(); // 服务实现类
        String methodName = request.getMethodName();    // 服务实现类的方法
        Class<?>[] parameterTypes = request.getParameterTypes(); // 参数类型
        Object[] parameters = {request.getData()}; // 参数
        //
        if (request.getData().getClass().isArray()){
            parameters = (Object[]) parameters[0];
        }

        // 提供对类的快速访问和调用
        FastClass fastClass = FastClass.create(serviceClass);
        // 获取到的指定方法在目标类中的索引值，可以用于后续的快速方法调用
        int methodIndex = fastClass.getIndex(methodName, parameterTypes);

        // 调用方法并返回结果
        return fastClass.invoke(methodIndex, serviceBean, parameters);
    }

}
